package com.api.apigateway.filter;

import cn.hutool.core.lang.UUID;
import cn.hutool.json.JSONUtil;
import com.api.apiclientsdk.utils.SignUtils;
import com.api.apigateway.mq.CompensateMessageProducer;
import com.api.common.alert.CommonAlertHelper;
import com.api.common.common.ErrorCode;
import com.api.common.common.ResultUtils;
import com.api.common.manager.APIInterfaceLogManager;
import com.api.common.model.dto.InvokeCountCompensation;
import com.api.common.model.entity.InterfaceInfo;
import com.api.common.model.entity.InterfaceInfoLog;
import com.api.common.model.entity.User;
import com.api.common.service.InnerInterfaceInfoService;
import com.api.common.service.InnerUserInterfaceInfoService;
import com.api.common.service.InnerUserService;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.reactivestreams.Publisher;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;

/**
 * @author: 崔老爷
 * Date: 2024/9/28 12:52
 * @Description: 全局过滤
 */

@Component
@Slf4j
public class ApiGlobalFilter implements GlobalFilter, Ordered {

    public static final String ATTR_START_NANOS="risk.startNanos";
    public static final String ATTR_CREATE_AT="risk.createAt";
    public static final String ATTR_REQUEST_ID="risk.requestId";
    public static final String ATTR_USER_ID="risk.userId";


    @DubboReference
    private InnerUserService innerUserService;
    @DubboReference
    private InnerInterfaceInfoService innerInterfaceInfoService;
    @DubboReference
    private InnerUserInterfaceInfoService innerUserInterfaceInfoService;
    @Resource
    private APIInterfaceLogManager apiInterfaceLogManager;
    @Resource
    private RedisTemplate<String,String> redisTemplate;
    @Resource
    private CompensateMessageProducer compensateMessageProducer;
    @Resource
    private CommonAlertHelper alertHelper;

    private static final String NONCE_PREFIX="api:gateway:nonce:";
    private static final Duration NONCE_EXPIRE = Duration.ofMinutes(10);
    private static final int MIN_NONCE_LENGTH = 16;
    /**
     * 黑白名单
     * 例如，如果某个远程地址频繁访问，我们可以将其添加到黑名单并拒绝访问。
     * 现在我们来试试设置一个规则，如果请求的来源地址不是 127.0.0.1，就拒绝它的访问。
     * ========
     * 先写一个全局的常量。在这里我们用一个白名单，通常建议在权限管理中尽量使用白名单，少用黑名单。
     * 白名单的原则是只允许特定的调用，这样可能会更加安全，或者你可以默认情况下全禁止。
     */
    private static final List<String> IP_WHITE_LIST = Arrays.asList("127.0.0.1");

    private static final String INTERFACE_INFO_PATH = "http://localhost:8100";



    /**
     * 1.用户发送请求到 API 网关
     *
     * @param exchange 路由交换机，我们所有的请求的信息、响应的信息、响应体、请求体都能从这里拿到。
     * @param chain    责任链模式，因为我们的所有过滤器是按照从上到下的顺序依次执行，形成了一个链条。所以这里用了一个chain，
     *                 如果当前过滤器对请求进行了过滤后发现可以放行，就要调用责任链中的next方法，相当于直接找到下一个过滤器，
     *                 这里称为filter。有时候我们需要在责任链中使用 next，而在这里它使用了 filter 来找到下一个过滤器，从而正常地放行请求。
     * @return Mono 其实是响应式编程的一种对象，类似于前端的 Promise
     */
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        log.info("-----------------ApiGlobalFilter-----------------");

        exchange.getAttributes().put(ATTR_START_NANOS,System.nanoTime());
        //createAt请求到网关的时间（毫秒时间戳）
        exchange.getAttributes().put(ATTR_CREATE_AT, System.currentTimeMillis());
        //requestId：优先复用上游传入的 X-Request-Id；没有就生成
        String incoming=exchange.getRequest().getHeaders().getFirst("X-Request-Id");
        String requestId=(StringUtils.isEmpty(incoming))? UUID.randomUUID().toString():incoming;
        exchange.getAttributes().put(ATTR_REQUEST_ID,requestId);

        //2.请求日志
        ServerHttpRequest request = exchange.getRequest();
        String path = INTERFACE_INFO_PATH + request.getPath().value();
        String method = request.getMethod().toString();
        String sourceAddress = Objects.requireNonNull(request.getLocalAddress()).getHostString();
        log.info("请求唯一标识：{}\n请求路径：{}\n请求方法：{}\n请求来源：{}",
                requestId,
                request.getPath(),
                method,
                maskIp(Optional.ofNullable(request.getRemoteAddress())
                        .map(addr -> addr.getAddress().getHostAddress())
                        .orElse("unknown"))
        );

        ServerHttpResponse response = exchange.getResponse();
        //3.访问控制-（黑白名单）
        //直接返回了这个 Mono，它并不包含响应参数。相当于我们告诉程序，请求处理完成了，不需要再执行其他操作了。
//        if (!IP_WHITE_LIST.contains(sourceAddress)) {
//            //不在白名单内的地址，就让其访问
//            response.setStatusCode(HttpStatus.FORBIDDEN);
//            return response.setComplete();
//        }
        //4.用户鉴权（判断 ak、sk 是否合法）
        HttpHeaders headers = request.getHeaders();
        String accessKey = headers.getFirst("accessKey");
        String nonce = headers.getFirst("nonce");
        String timestamp = headers.getFirst("timestamp");
        String sign = headers.getFirst("sign");
        String body = headers.getFirst("body");

        // 直接校验如果随机数大于1万，则抛出异常，并提示"无权限"
        if(!validNonce(nonce,timestamp)){
            String ip = Optional.ofNullable(request.getRemoteAddress())
                    .map(addr -> addr.getAddress().getHostAddress())
                    .orElse("unknown");
            alertHelper.suspiciousIp(ip, "检测到重放攻击", 1);
            return handleNoAuth(response);
        }

        // 时间和当前时间不能超过 5 分钟
        // 首先,获取当前时间的时间戳,以秒为单位
        // System.currentTimeMillis()返回当前时间的毫秒数，除以1000后得到当前时间的秒数。
        Long currentTime = System.currentTimeMillis() / 1000;
        // 定义一个常量FIVE_MINUTES,表示五分钟的时间间隔(乘以60,将分钟转换为秒,得到五分钟的时间间隔)。
        final Long FIVE_MINUTES = 60 * 5L;
        // 判断当前时间与传入的时间戳是否相差五分钟或以上
        // Long.parseLong(timestamp)将传入的时间戳转换成长整型
        // 然后计算当前时间与传入时间戳之间的差值(以秒为单位),如果差值大于等于五分钟,则返回true,否则返回false
        if ((currentTime - Long.parseLong(timestamp)) >= FIVE_MINUTES) {
            // 如果时间戳与当前时间相差五分钟或以上，调用handleNoAuth(response)方法进行处理
            return handleNoAuth(response);
        }

        User invokeUser = null;
        try {
            invokeUser = innerUserService.getInvokeUser(accessKey);
            exchange.getAttributes().put(ATTR_USER_ID,invokeUser.getId());
        } catch (Exception e) {
            log.error("获取用户信息失败", e);
        }
        if (invokeUser == null) {
            return handleNoAuth(response);
        }

        // 实际情况中是从数据库中查出 secretKey
        String secretKey = invokeUser.getSecretKey();
        String serverSign = SignUtils.genSign(body, secretKey);
        if (sign == null || !sign.equals(serverSign)) {
            return handleNoAuth(response);
        }
        //5.请求的模拟接口是否存在？
        //从数据库中查询模拟接口是否存在，已经请求方法是否匹配,这边调用前端直接传入参数和对应接口id
        InterfaceInfo interfaceInfo = null;
        Gson gson = new Gson();
        if (StringUtils.isBlank(body)) {
            return handleNoAuth(response);
        }
        Map<String, String> bodyMap = gson.fromJson(body, Map.class);
        //todo 这边接口id只适合前端的测试调用，如果sdk方式应该使用url+method的方式
        String id = bodyMap.get("id");
        try {
//            interfaceInfo = innerInterfaceInfoService.getInterfaceInfo(path, method);
            interfaceInfo = innerInterfaceInfoService.getInterfaceInfoById(id);
        } catch (Exception e) {
            log.error("getInterfaceInfoById error: {}", e.getMessage());
        }
        if (interfaceInfo == null) {
            return handleNoAuth(response);
        }
        final Long userId = invokeUser.getId();
        final Long interfaceInfoId = interfaceInfo.getId();
        //判断调用接口是否是测试调用，如果是测试调用

        /**
         * todo 用户是否还有调用次数
         * 根据用户id和接口id获取用户接口关系详情对象
         * 从用户接口关系详情对象中获取剩余调用次数
         * 判断剩余调用次数是否 >0 如果小于0则直接拒绝
         * 大于0则直接执行调用接口流程
         * 因为是在网关层统一实现，为避免在网关层引入业务层的环境，加深耦合度，我们采取远程RPC调用接口管理平台的接口实现
         */
        int invokeInterfaceNum = innerUserInterfaceInfoService.invokeInterfaceNum(userId, interfaceInfoId);
        if (invokeInterfaceNum <= 0) {
            //当接口数量耗尽时，进行告警
            String email = invokeUser.getEmail();
            if(StringUtils.isNotBlank(email))
                alertHelper.interfaceQuotaExhausted(userId, interfaceInfoId,  interfaceInfo.getName(), invokeInterfaceNum,email);
            return handleNoInterfaceInvokeCount(response);
        }
        //6.请求转发，调用模拟接口 + 7.响应日志 + 8.调用成功，接口调用次数 + 1
        return handleResponse(exchange, chain, interfaceInfoId, userId);

    }

    /**
     * 校验nonce是否已经被使用（防重放攻击）
     *
     * @param nonce
     * @param timestamp
     * @return false 被使用， true未被使用
     */
    private boolean validNonce(String nonce, String timestamp) {
        if (StringUtils.isBlank(nonce)||nonce.length()<MIN_NONCE_LENGTH) {
            log.warn("Nonce格式不正确: length={}", nonce == null ? 0 : nonce.length());
            return false;
        }
        String cachedKey=NONCE_PREFIX+nonce;
        Boolean bool = redisTemplate.opsForValue()
                .setIfAbsent(cachedKey, timestamp, NONCE_EXPIRE);

        if(!Boolean.TRUE.equals(bool)){
            log.warn("检测到重复攻击：nonce={}",nonce);
            return false;
        }
        return true;
    }

    /**
     * 处理响应
     *
     * @param exchange
     * @param chain
     * @return
     */
    public Mono<Void> handleResponse(ServerWebExchange exchange, GatewayFilterChain chain, long interfaceInfoId, long userId) {
        try {
            // 获取原始的响应对象
            ServerHttpResponse originalResponse = exchange.getResponse();
            // 缓存数据的工厂
            DataBufferFactory bufferFactory = originalResponse.bufferFactory();
            // 拿到响应码
            HttpStatus statusCode = originalResponse.getStatusCode();

            // 从 exchange 中获取 requestId
            String requestId = (String) exchange.getAttributes().get(ATTR_REQUEST_ID);

            // 判断状态码是否为200 OK(按道理来说,现在没有调用,是拿不到响应码的,对这个保持怀疑 沉思.jpg)
            if (statusCode == HttpStatus.OK) {
                // 创建一个装饰后的响应对象(开始穿装备，增强能力)
                ServerHttpResponseDecorator decoratedResponse = new ServerHttpResponseDecorator(originalResponse) {
                    // 重写writeWith方法，用于处理响应体的数据
                    // 这段方法就是只要当我们的模拟接口调用完成之后,等它返回结果，
                    // 就会调用writeWith方法,我们就能根据响应结果做一些自己的处理
                    @Override
                    public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                        log.info("body instanceof Flux: {}", (body instanceof Flux));
                        if (body instanceof Flux) {
                            Flux<? extends DataBuffer> fluxBody = Flux.from(body);
                            // 返回一个处理后的响应体
                            // (这里就理解为它在拼接字符串,它把缓冲区的数据取出来，一点一点拼接好)
                            return super.writeWith(
                                    fluxBody.map(dataBuffer -> {
                                        // 8. 调用成功，接口调用次数 + 1 invokeCount
                                        try {
                                            boolean invoked = innerUserInterfaceInfoService.invokeCount(interfaceInfoId, userId);
                                            if(invoked)log.info("调用扣款成功, interfaceInfoId:{}, userId:{}, requestId={}", interfaceInfoId, userId,requestId);
                                        } catch (Exception e) {
                                            log.error("调用扣款失败, interfaceInfoId:{}, userId:{}", interfaceInfoId, userId, e);

                                            // 发送补偿消息到MQ（由主服务消费）
                                            sendCompensationMessage(interfaceInfoId, userId, requestId, e.getMessage());
                                        }
                                        // 读取响应体的内容并转换为字节数组
                                        byte[] content = new byte[dataBuffer.readableByteCount()];
                                        dataBuffer.read(content);
                                        DataBufferUtils.release(dataBuffer);//释放掉内存
                                        // 构建日志
                                        StringBuilder sb2 = new StringBuilder(200);
                                        List<Object> rspArgs = new ArrayList<>();
                                        rspArgs.add(originalResponse.getStatusCode());
                                        //这里的 data 就是真实的响应结果，这个 data 是从 content 拿的。
                                        //content 又从哪里来？就是从我们响应的数据流中取到的。(实际上就是我们原本模拟接口返回的数据)
                                        String data = new String(content, StandardCharsets.UTF_8); //data
                                        sb2.append(data);
                                        // 打印日志
                                        log.info("响应结果：{}", data);
                                        // 将处理后的内容重新包装成DataBuffer并返回
                                        return bufferFactory.wrap(content);
                                    }));
                        } else {
                            //9.调用失败，返回一个规范的错误码
                            log.error("<--- {} 响应code异常", getStatusCode());
                        }
                        return super.writeWith(body);
                    }
                };
                sendInterfaceInfoLog(interfaceInfoId, userId, decoratedResponse.getStatusCode().value(), null);
                // 对于200 OK的请求,将装饰后的响应对象传递给下一个过滤器链,并继续处理(设置repsonse对象为装饰过的)
                return chain.filter(exchange.mutate().response(decoratedResponse).build());
            }
            // 对于非200 OK的请求，直接返回，进行降级处理
            return chain.filter(exchange);
        } catch (Exception e) {
            log.error("网关处理响应异常" , e);
            // ⚠️ 网关核心异常告警：
            alertHelper.gatewayRouteFailure(
                    String.valueOf(interfaceInfoId),
                    exchange.getRequest().getPath().value(),
                    e.getMessage()
            );
            sendInterfaceInfoLog(interfaceInfoId, userId, e.hashCode(), e.getMessage());
            return chain.filter(exchange);
        }
    }

    private void sendInterfaceInfoLog(long interfaceInfoId, long userId, Integer code, String message) {
        InterfaceInfoLog interfaceInfoLog = new InterfaceInfoLog();
        interfaceInfoLog.setInterface_info_id(interfaceInfoId);
        interfaceInfoLog.setUser_id(userId);
        interfaceInfoLog.setCode(code);
        interfaceInfoLog.setMessage(code == 200 ? "OK" : message);
        Gson gson = new Gson();
        apiInterfaceLogManager.sendMessage(gson.toJson(interfaceInfoLog));
    }
    /**
     * 发送补偿消息到MQ
     */
    private void sendCompensationMessage(Long interfaceInfoId, Long userId, String requestId, String errorMessage) {
        try {
            InvokeCountCompensation compensation = new InvokeCountCompensation();
            compensation.setInterfaceInfoId(interfaceInfoId);
            compensation.setUserId(userId);
            compensation.setRetryCount(0);
            compensation.setCreateTime(System.currentTimeMillis());
            compensation.setErrorMessage(errorMessage);
            compensation.setRequestId(requestId);

            Gson gson = new Gson();
            compensateMessageProducer.send(gson.toJson(compensation));
        } catch (Exception e) {
            log.error("发送补偿消息失败: interfaceInfoId={}, userId={}, requestId={}",
                    interfaceInfoId, userId, requestId, e);
        }
    }

    /**
     * 返回接口没有调用次数时，返回错误信息
     *
     * @param response
     * @return
     */
    public Mono<Void> handleNoInterfaceInvokeCount(ServerHttpResponse response) {
        // 当请求发生错误时自定义返回值
        response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
        response.getHeaders().add(HttpHeaders.CONTENT_TYPE, "application/json");
        byte[] bytes = JSONUtil.toJsonStr(ResultUtils.error(ErrorCode.INVOKE_INTERFACE_ERROR, "接口调用次数已用尽")).getBytes();
        // 使用 response.writeWith 将数据写入响应体
        /**
         * response.writeWith：这个方法接受一个 Publisher<DataBuffer>，将数据写入响应的 body。这里通过 Mono.just(buffer) 来将 DataBuffer 写入响应。
         * response.writeWith 已经完成了响应体的写入，因此不需要再调用 response.setComplete()。
         * DataBuffer：使用 response.bufferFactory().wrap(bytes) 将字节数组包装为 DataBuffer。
         */
        DataBuffer buffer = response.bufferFactory().wrap(bytes);
        response.getHeaders().setContentLength(bytes.length);
        return response.writeWith(Mono.just(buffer));
    }

    public Mono<Void> handleNoAuth(ServerHttpResponse response) {
        response.setStatusCode(HttpStatus.FORBIDDEN);
        return response.setComplete();
    }

    public Mono<Void> handleInvokeError(ServerHttpResponse response) {
        response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
        return response.setComplete();
    }

    /**
     * 值越低优先级越高
     * 设置为 0，确保在限流过滤器（Order 通常为 -1）之后执行
     * 
     * @return
     */
    @Override
    public int getOrder() {
        return 0;
    }


    // 脱敏工具
    private String maskIp(String ip) {
        if (ip == null) return "unknown";
        String[] parts = ip.split("\\.");
        if (parts.length == 4) {
            return parts[0] + "." + parts[1] + ".*.*";
        }
        return "***";
    }
}
